package qy.jalgotrade.utils;

import java.io.*;
import java.nio.charset.Charset;
import java.nio.file.Path;
import java.time.ZoneId;
import java.util.*;

import com.google.common.collect.HashBasedTable;
import org.apache.commons.collections4.IteratorUtils;
import org.apache.commons.collections4.OrderedMap;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.csv.CSVRecord;
import org.apache.commons.io.output.FileWriterWithEncoding;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.math3.stat.StatUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.math.DoubleMath;

import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import jep.Interpreter;
import jep.JepException;
import jep.python.PyCallable;
import jep.python.PyObject;
import tech.tablesaw.api.*;
import tech.tablesaw.columns.Column;
import tech.tablesaw.io.DataFrameWriter;
import tech.tablesaw.io.csv.CsvReadOptions;
import tech.tablesaw.io.csv.CsvReadOptions.Builder;

/**
 * @author c-geo
 *
 */
public final class CommonUtils {

	private static final Logger logger = LoggerFactory.getLogger(CommonUtils.class);

	/**
	 * http://en.wikipedia.org/wiki/List_of_market_opening_times
	 * 
	 * @author c-geo
	 *
	 */
	public enum MarketSession {
		// @formatter:off
		NASDAQ("US/Eastern"), // NASDAQ market session.
		NYSE("US/Eastern"), // New York Stock Exchange market session.
		USEquities("US/Eastern"), // US Equities market session.
		MERVAL("America/Argentina/Buenos_Aires"), // Buenos Aires (Argentina) market session.
		BOVESPA("America/Sao_Paulo"), // BOVESPA (Brazil) market session.
		FTSE("Europe/London"), // London Stock Exchange market session.
		TSE("Asia/Tokyo"), // Tokyo Stock Exchange market session.
		SSE("Asia/Shanghai"); // Shanghai Stock Exchange market session.
		// @formatter:on

		private final String value;

		private final ZoneId zoneId;

		private MarketSession(String value) {

			this.value = value;
			zoneId = ZoneId.of(value);
		}

		public String getValue() {

			return value;
		}

		public ZoneId getZoneId() {

			return zoneId;
		}
	}

	public static final String NUMBER_OF_PROCESSORS = "NUMBER_OF_PROCESSORS";

	public static double FUZZY_EQUAL_TOLERANCE_MATH = 0.000000000001;

	public static double FUZZY_EQUAL_TOLERANCE_ASSET = 0.000000001;

	/**
	 * {@link CSVFormat CSVFormat.EXCEL.withFirstRecordAsHeader().withIgnoreEmptyLines()}
	 */
	public static CSVFormat CSV_FORMAT_DEFAULT = CSVFormat.EXCEL.withFirstRecordAsHeader().withIgnoreEmptyLines();

	/**
	 * return DoubleMath.fuzzyEquals(lVal, rVal, FUZZY_EQUAL_TOLERANCE_MATH);
	 * 
	 * @param lVal lVal
	 * @param rVal rVal
	 * @return boolean
	 */
	public static boolean fuzzyEquals(double lVal, double rVal) {

		return DoubleMath.fuzzyEquals(lVal, rVal, FUZZY_EQUAL_TOLERANCE_MATH);
	}

	/**
	 *
	 * @param val val
	 * @return boolean
	 */
	public static boolean fuzzyEqualZero(double val) {

		return DoubleMath.fuzzyEquals(val, 0.0, FUZZY_EQUAL_TOLERANCE_MATH);
	}

	/**
	 * return DoubleMath.fuzzyCompare(lVal, rVal, FUZZY_EQUAL_TOLERANCE_MATH);
	 * 
	 * @param lVal lVal
	 * @param rVal rVal
	 * @return boolean
	 */
	public static int fuzzyCompare(double lVal, double rVal) {

		return DoubleMath.fuzzyCompare(lVal, rVal, FUZZY_EQUAL_TOLERANCE_MATH);
	}

	/**
	 * 仅当 left 和 right 均非 None 时, 返回 min(left, right); 当 left 和 right 均为 None 时返回 None:
	 * 
	 * @param left left
	 * @param right right
	 * @return boolean
	 */
	@SuppressWarnings("unchecked")
	public static <T> T safeMin(Comparable<T> left, Comparable<T> right) {

		if (left == null) {
			return (T) right;
		} else if (right == null) {
			return (T) left;
		} else {
			if (left.compareTo((T) right) < 0) {
				return (T) left;
			} else {
				return (T) right;
			}
		}
	}

	/**
	 * 仅当 left 和 right 均非 None 时, 返回 max(left, right); 当 left 和 right 均为 None 时返回 None:
	 * 
	 * @param left left
	 * @param right right
	 * @return T
	 */
	@SuppressWarnings("unchecked")
	public static <T> T safeMax(Comparable<T> left, Comparable<T> right) {

		if (left == null) {
			return (T) right;
		} else if (right == null) {
			return (T) left;
		} else {
			if (left.compareTo((T) right) > 0) {
				return (T) left;
			} else {
				return (T) right;
			}
		}
	}

	/**
	 * return (actual - prev) / Math.abs(prev);
	 * 
	 * @param actual actual
	 * @param prev prev
	 * @return pctChange
	 */
	public static double getChangePercentage(double actual, double prev) {

		if (Double.isNaN(actual) || Double.isNaN(prev) || DoubleMath.fuzzyEquals(prev, 0, FUZZY_EQUAL_TOLERANCE_MATH)) {
			throw new IllegalArgumentException(String.format("Invalid values (actual: %s, prev: %s)", actual, prev));
		}

		double diff = actual - prev;
		return diff / Math.abs(prev);
	}

	/**
	 * 
	 * @param open open
	 * @param high high
	 * @param low low
	 * @param close close
	 * @return [open, high, low, close]
	 */
	public static List<Double> sanitizeOhlc(double open, double high, double low, double close) {

		if (low > open) {
			low = open;
		}
		if (low > close) {
			low = close;
		}
		if (high < open) {
			high = open;
		}
		if (high < close) {
			high = close;
		}
		return Arrays.asList(open, high, low, close);
	}

	/**
	 * 
	 * @param da da
	 * @return max
	 */
	public static double max(double[] da) {

		return StatUtils.max(da);
	}

	/**
	 * 
	 * @param dal dal
	 * @return max
	 */
	public static double max(DoubleArrayList dal) {

		return StatUtils.max(dal.toDoubleArray());
	}

	/**
	 * 
	 * @param da da
	 * @return min
	 */
	public static double min(double[] da) {

		return StatUtils.min(da);
	}

	/**
	 * 
	 * @param dal dal
	 * @return min
	 */
	public static double min(DoubleArrayList dal) {

		return StatUtils.min(dal.toDoubleArray());
	}

	/**
	 * 
	 * @param da da
	 * @return sum
	 */
	public static double sum(double[] da) {

		return StatUtils.sum(da);
	}

	/**
	 * 
	 * @param dal dal
	 * @return sum
	 */
	public static double sum(DoubleArrayList dal) {

		return StatUtils.sum(dal.toDoubleArray());
	}

	/**
	 * 
	 * @param da da
	 * @return mean
	 */
	public static double mean(double[] da) {

		return StatUtils.mean(da);
	}

	/**
	 * 
	 * @param dal dal
	 * @return mean
	 */
	public static double mean(DoubleArrayList dal) {

		return StatUtils.mean(dal.toDoubleArray());
	}

	/**
	 * 
	 * @param da da
	 * @return variance
	 */
	public static double variance(double[] da) {

		return StatUtils.variance(da);
	}

	/**
	 * 
	 * @param dal dal
	 * @return variance
	 */
	public static double variance(DoubleArrayList dal) {

		return StatUtils.variance(dal.toDoubleArray());
	}

	/**
	 * 
	 * @param da da
	 * @return std
	 */
	public static double std(double[] da) {

		return Math.sqrt(variance(da));
	}

	/**
	 * 
	 * @param dal dal
	 * @return std
	 */
	public static double std(DoubleArrayList dal) {

		return std(dal.toDoubleArray());
	}

	/**
	 * 
	 * @param da da
	 * @return populationVariance
	 */
	public static double populationVariance(double[] da) {

		return StatUtils.populationVariance(da);
	}

	/**
	 * 
	 * @param dal dal
	 * @return populationVariance
	 */
	public static double populationVariance(DoubleArrayList dal) {

		return StatUtils.populationVariance(dal.toDoubleArray());
	}

	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	// Jep helpers:
	////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
	/**
	 * 
	 * @param className className
	 * @param methodName methodName
	 * @param udName udName
	 * @return PyObjectName
	 */
	public static String genPyObjectName(String className, String methodName, String udName) {

		return String.format("%s_%s_%s", className, methodName, udName).replaceAll("[^\\w]", "_");
	}

	/**
	 *
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @return jepInterp
	 */
	public static boolean pyObjectExists(Interpreter jepInterp, String pyObjName) {

		try {
			return jepInterp.getValue(pyObjName) != null;
		} catch (JepException e) {
			return false;
		}
	}

	/**
	 * Jep Interpreter 对象命名规范: &lt;Class.getName()&gt; "_" &lt;method-name&gt; "_" &lt;...&gt;
	 * 
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @return PyObject
	 */
	public static PyObject createPyList(Interpreter jepInterp, String pyObjName) {

		return createPyList(jepInterp, pyObjName, false);
	}

	/**
	 *
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @param dropExisting dropExisting
	 * @return PyObject
	 */
	public static PyObject createPyList(Interpreter jepInterp, String pyObjName, boolean dropExisting) {

		try {
			boolean pyLsExisting = pyObjectExists(jepInterp, pyObjName);
			if (dropExisting || !pyLsExisting) {
				jepInterp.exec(String.format("%s = []", pyObjName));
				return jepInterp.getValue(pyObjName, PyObject.class);
			} else {
				logger.error("python list: {} already existed !", pyObjName);
				return null;
			}
		} catch (JepException e) {
			logger.error("Exception occurred creating python list: {}", ExceptionUtils.getStackTrace(e));
			return null;
		}
	}

	/**
	 *
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @return PyObject
	 */
	public static PyObject createOrGetPyList(Interpreter jepInterp, String pyObjName) {

		try {
			jepInterp.exec(String.format("%s = [] if \"%s\" not in locals() else %s", pyObjName, pyObjName, pyObjName));
			return jepInterp.getValue(pyObjName, PyObject.class);
		} catch (JepException e) {
			logger.error("Exception occurred creating python list: {}", ExceptionUtils.getStackTrace(e));
			return null;
		}
	}

	/**
	 * Jep Interpreter 对象命名规范: &lt;Class.getName()&gt; "_" &lt;method-name&gt; "_" &lt;...&gt;
	 * 
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @return PyObject
	 */
	public static PyObject createPyDict(Interpreter jepInterp, String pyObjName) {

		return createPyDict(jepInterp, pyObjName, false);
	}

	/**
	 *
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @param dropExisting dropExisting
	 * @return PyObject
	 */
	public static PyObject createPyDict(Interpreter jepInterp, String pyObjName, boolean dropExisting) {

		try {
			boolean pyDictExisting = pyObjectExists(jepInterp, pyObjName);
			if (dropExisting || !pyDictExisting) {
				jepInterp.exec(String.format("%s = {}", pyObjName));
				return jepInterp.getValue(pyObjName, PyObject.class);
			} else {
				logger.error("python dict: {} already existed !", pyObjName);
				return null;
			}
		} catch (JepException e) {
			logger.error("Exception occurred creating python dict: {}", ExceptionUtils.getStackTrace(e));
			return null;
		}
	}

	/**
	 *
	 * @param jepInterp jepInterp
	 * @param pyObjName pyObjName
	 * @return PyObject
	 */
	public static PyObject createOrGetPyDict(Interpreter jepInterp, String pyObjName) {

		try {
			jepInterp.exec(String.format("%s = {}} if \"%s\" not in locals() else %s", pyObjName, pyObjName, pyObjName));
			return jepInterp.getValue(pyObjName, PyObject.class);
		} catch (JepException e) {
			logger.error("Exception occurred creating python dict: {}", ExceptionUtils.getStackTrace(e));
			return null;
		}
	}

	/**
	 *
	 * @param jepInterp jepInterp
	 * @param constructorName constructorName
	 * @param pyObjName pyObjName
	 * @param args args
	 * @param kwargs kwargs
	 * @return PyObject
	 * @throws JepException JepException
	 */
	public static PyObject createPyObjectBy(Interpreter jepInterp, String constructorName, String pyObjName,
											Object[] args, Map<String, Object> kwargs) throws JepException {

		PyCallable pyConstructor = jepInterp.getValue(constructorName, PyCallable.class);
		PyObject pyObj = pyConstructor.callAs(PyObject.class, args, kwargs);
		jepInterp.set(pyObjName, pyObj);
		return pyObj;
	}

	/**
	 * 
	 * @param pyObj pyObj
	 * @param methodName methodName
	 * @param args args
	 * @param kwargs kwargs
	 * @throws JepException JepException
	 */
	public static Object pyInvoke(PyObject pyObj, String methodName, Object[] args, Map<String, Object> kwargs)
	        throws JepException {

		PyCallable method = pyObj.getAttr(methodName, PyCallable.class);
		return method.call(args, kwargs);
	}

	/**
	 * 
	 * @param pyObj pyObj
	 * @param methodName methodName
	 * @param args args
	 * @return Object
	 * @throws JepException JepException
	 */
	public static Object pyInvoke(PyObject pyObj, String methodName, Object... args) throws JepException {

		PyCallable method = pyObj.getAttr(methodName, PyCallable.class);
		return method.call(args);
	}

	/**
	 * 
	 * @param pyObj pyObj
	 * @param methodName methodName
	 * @param kwargs kwargs
	 * @return Object
	 * @throws JepException JepException
	 */
	public static Object pyInvoke(PyObject pyObj, String methodName, Map<String, Object> kwargs) throws JepException {

		PyCallable method = pyObj.getAttr(methodName, PyCallable.class);
		return method.call(kwargs);
	}

	/**
	 * 
	 * @param expectedType expectedType
	 * @param pyObj pyObj
	 * @param methodName methodName
	 * @param args args
	 * @param kwargs kwargs
	 * @return T
	 * @throws JepException JepException
	 */
	public static <T> T pyInvokeAs(Class<T> expectedType, PyObject pyObj, String methodName, Object[] args,
	        Map<String, Object> kwargs) throws JepException {

		PyCallable method = pyObj.getAttr(methodName, PyCallable.class);
		return method.callAs(expectedType, args, kwargs);
	}

	/**
	 * 
	 * @param expectedType expectedType
	 * @param pyObj pyObj
	 * @param methodName methodName
	 * @param args args
	 * @return T
	 * @throws JepException JepException
	 */
	public static <T> T pyInvokeAs(Class<T> expectedType, PyObject pyObj, String methodName, Object... args)
	        throws JepException {

		PyCallable method = pyObj.getAttr(methodName, PyCallable.class);
		return method.callAs(expectedType, args);
	}

	/**
	 * 
	 * @param expectedType expectedType
	 * @param pyObj pyObj
	 * @param methodName methodName
	 * @param kwargs kwargs
	 * @return T
	 * @throws JepException JepException
	 */
	public static <T> T pyInvokeAs(Class<T> expectedType, PyObject pyObj, String methodName, Map<String, Object> kwargs)
	        throws JepException {

		PyCallable method = pyObj.getAttr(methodName, PyCallable.class);
		return method.callAs(expectedType, kwargs);
	}

	/**
	 * 获取 csv 列信息:
	 * 
	 * @param csvPath csvPath
	 * @return The map keys are column names. The map values are 0-based indices.
	 * @throws IOException IOException
	 */
	public static Map<String, Integer> csvHeaderMap(Path csvPath) throws IOException {

		return csvHeaderMap(csvPath, "UTF-8");
	}

	/**
	 * 获取 csv 列信息:
	 *
	 * @param csvIn csvIn
	 * @return The map keys are column names. The map values are 0-based indices.
	 * @throws IOException IOException
	 */
	public static Map<String, Integer> csvHeaderMap(InputStream csvIn) throws IOException {

		return csvHeaderMap(csvIn, "UTF-8");
	}

	/**
	 * 获取 csv 列信息:
	 * 
	 * @param csvPath csvPath
	 * @return The map keys are column names. The map values are 0-based indices.
	 * @throws IOException IOException
	 */
	public static Map<String, Integer> csvHeaderMap(Path csvPath, String enc) throws IOException {

		try (InputStream csvIn = new FileInputStream(csvPath.toFile())) {
			return csvHeaderMap(csvIn, enc);
		}
	}

	/**
	 * 获取 csv 列信息:
	 *
	 * @param csvIn csvIn
	 * @param enc enc
	 * @return The map keys are column names. The map values are 0-based indices.
	 * @throws IOException IOException
	 */
	public static Map<String, Integer> csvHeaderMap(InputStream csvIn, String enc) throws IOException {

		CSVParser parser = CSVParser.parse(csvIn, Charset.forName(enc),
				CSVFormat.DEFAULT.withFirstRecordAsHeader().withAllowMissingColumnNames());
		return parser.getHeaderMap();
	}

	/**
	 * tablesaw (https://github.com/jtablesaw/tablesaw) 暂不支持 (0.38.0) 指定需要类型的列子集:
	 * 
	 * @param csvPath 预读取 csv 的完整 header
	 * @param desired 需要类型的列子集
	 * @return {@link ColumnType ColumnType[]} 可作为 {@link Builder#columnTypes(ColumnType[])} 的参数.
	 * @throws IOException IOException
	 */
	public static ColumnType[] genColumnsInfoFor(Path csvPath, Map<String, ColumnType> desired) throws IOException {

		return genColumnsInfoFor(csvPath, desired, "UTF-8");
	}

	/**
	 * tablesaw (https://github.com/jtablesaw/tablesaw) 暂不支持 (0.38.0) 指定需要类型的列子集:
	 * 
	 * @param csvPath 预读取 csv 的完整 header
	 * @param desired 需要类型的列子集
	 * @param enc enc
	 * @return {@link ColumnType ColumnType[]} 可作为 {@link Builder#columnTypes(ColumnType[])} 的参数.
	 * @throws IOException IOException
	 */
	public static ColumnType[] genColumnsInfoFor(Path csvPath, Map<String, ColumnType> desired, String enc)
	        throws IOException {

		Map<String, Integer> allHeaders = csvHeaderMap(csvPath, enc);
		String[] headerNames = new String[allHeaders.size()];
		for (Map.Entry<String, Integer> head : allHeaders.entrySet()) {
			headerNames[head.getValue()] = head.getKey();
		}

		ColumnType[] ret = new ColumnType[allHeaders.size()];
		for (String e : desired.keySet()) {
			int colIdx;
			if (!allHeaders.containsKey(e)) {
				throw new IllegalArgumentException(
				        String.format("desired Column (%s) does NOT exists: %s", e, csvPath));
			} else {
				colIdx = allHeaders.get(e);
			}
			ret[colIdx] = desired.get(e);
		}
		for (int i = 0; i < allHeaders.size(); i++) {
			if (ret[i] == null) {
				ret[i] = ColumnType.SKIP;
			}
		}
		return ret;
	}

	/**
	 * 创建新 {@link Table}, 所有列类型都为 String.
	 * @param colNames colNames
	 * @return Table
	 */
	public static Table makeTable(List<String> colNames) {

		List<Column<?>> cols = new ArrayList<>();
		for (String e : colNames) {
			cols.add(StringColumn.create(e));
		}
		return Table.create(cols);
	}

	/**
	 * 创建新 {@link Table}.
	 * 
	 * @param colInfos colInfos
	 * @return Table
	 */
	public static Table makeTable(OrderedMap<String, ColumnType> colInfos) {

		List<Column<?>> cols = new ArrayList<>();
		for (Map.Entry<String, ColumnType> e : colInfos.entrySet()) {
			Column<?> col = e.getValue().create(e.getKey());
			cols.add(col);
		}
		return Table.create(cols);
	}

	/**
	 * 创建新 Table: 列元数据仿照源 Table:
	 *
	 * @param table table
	 * @return Table
	 */
	public static Table makeTableWithSameColumns(Table table) {

		if (table.columns().size() > 0) {
			Column<?>[] columns = new Column<?>[table.columns().size()];
			for (int i = 0; i < columns.length; i++) {
				Column<?> srcColumn = table.columns().get(i);
				columns[i] = srcColumn.type().create(srcColumn.name());
			}
			return Table.create(columns);
		} else {
			return Table.create();
		}
	}

	/**
	 *
	 * @param row Row
	 * @param columns columns
	 * @return a row
	 */
	public static Map<String, Double> tableRow2MapDouble(Row row, List<String> columns) {

		Map<String, Double> ret = new HashMap<>();
		for (String e : columns) {
			ret.put(e, row.getDouble(e));
		}
		return ret;
	}

	/**
	 *
	 * @param row row
	 * @param columnsP columnsP
	 * @return a row
	 */
	public static Map<String, String> tableRow2MapString(Row row, Optional<List<String>> columnsP) {

		List<String> columns = columnsP.orElseGet(row::columnNames);
		Map<String, String> ret = new HashMap<>();
		for (String e : columns) {
			ret.put(e, row.getString(e));
		}
		return ret;
	}

	/**
	 *
	 * @param csvPath csvPath
	 * @return Pair&lt;List&lt;String&gt;, List&lt;CSVRecord&gt;&gt;
	 * @throws IOException IOException
	 */
	public static Pair<List<String>, List<CSVRecord>> readCsvSimple(Path csvPath) throws IOException {

		return readCsvSimple(csvPath, "UTF-8");
	}

	/**
	 *
	 * @param csvIn csvIn
	 * @return Pair&lt;List&lt;String&gt;, List&lt;CSVRecord&gt;&gt;
	 * @throws IOException IOException
	 */
	public static Pair<List<String>, List<CSVRecord>> readCsvSimple(InputStream csvIn) throws IOException {

		return readCsvSimple(csvIn, "UTF-8");
	}

	/**
	 *
	 * @param csvPath csvPath
	 * @param enc default "UTF-8"
	 * @return Pair&lt;List&lt;String&gt;, List&lt;CSVRecord&gt;&gt;
	 * @throws IOException IOException
	 */
	public static Pair<List<String>, List<CSVRecord>> readCsvSimple(Path csvPath, String enc) throws IOException {

		Pair<List<String>, List<CSVRecord>> ret = Pair.of(Collections.emptyList(), Collections.emptyList());
		try (CSVParser parser = CSVParser.parse(csvPath, Charset.forName(enc),
				CSVFormat.DEFAULT.withFirstRecordAsHeader())) {
			List<String> headerNames = parser.getHeaderNames();
			List<CSVRecord> records = parser.getRecords();
			ret = Pair.of(headerNames, records);
		}
		return ret;
	}

	/**
	 *
	 * @param csvIn csvIn
	 * @param enc default "UTF-8"
	 * @return Pair&lt;List&lt;String&gt;, List&lt;CSVRecord&gt;&gt;
	 * @throws IOException IOException
	 */
	public static Pair<List<String>, List<CSVRecord>> readCsvSimple(InputStream csvIn, String enc) throws IOException {

		Pair<List<String>, List<CSVRecord>> ret = Pair.of(Collections.emptyList(), Collections.emptyList());
		CSVParser parser = CSVParser.parse(csvIn, Charset.forName(enc),
				CSVFormat.DEFAULT.withFirstRecordAsHeader());
		List<String> headerNames = parser.getHeaderNames();
		List<CSVRecord> records = parser.getRecords();
		ret = Pair.of(headerNames, records);

		return ret;
	}

	/**
	 *
	 * @param csvPath csvPath
	 * @param keyColumn keyColumn
	 * @return com.google.common.collect.Table
	 * @throws IOException IOException
	 */
	public static com.google.common.collect.Table<String, String, String> readCsv(Path csvPath, String keyColumn)
			throws IOException {

		return readCsv(csvPath, keyColumn, "UTF-8");
	}

	/**
	 *
	 * @param csvIn csvIn
	 * @param keyColumn keyColumn
	 * @return com.google.common.collect.Table
	 * @throws IOException IOException
	 */
	public static com.google.common.collect.Table<String, String, String> readCsv(InputStream csvIn, String keyColumn)
			throws IOException {

		return readCsv(csvIn, keyColumn, "UTF-8");
	}

	/**
	 *
	 * @param csvIn csvIn
	 * @param keyColumn keyColumn
	 * @param enc enc
	 * @return com.google.common.collect.Table
	 * @throws IOException IOException
	 */
	public static com.google.common.collect.Table<String, String, String> readCsv(InputStream csvIn, String keyColumn,
																				  String enc) throws IOException {

		Pair<List<String>, List<CSVRecord>> csvContentTup = readCsvSimple(csvIn, enc);
		if (!csvContentTup.getLeft().contains(keyColumn)) {
			throw new IllegalArgumentException(String.format("Key column: %s NOT found in csvIn: %s", keyColumn, csvIn));
		}
		HashBasedTable<String, String, String> ret = HashBasedTable.create();
		for (CSVRecord rec : csvContentTup.getRight()) {
			String rowKey = rec.get(keyColumn);
			for (String column : csvContentTup.getLeft()) {
				ret.put(rowKey, column, rec.get(column));
			}
		}
		return ret;
	}

	/**
	 *
	 * @param csvPath csvPath
	 * @param keyColumn keyColumn
	 * @param enc enc
	 * @return com.google.common.collect.Table
	 * @throws IOException IOException
	 */
	public static com.google.common.collect.Table<String, String, String> readCsv(Path csvPath, String keyColumn,
																				  String enc) throws IOException {

		Pair<List<String>, List<CSVRecord>> csvContentTup = readCsvSimple(csvPath, enc);
		if (!csvContentTup.getLeft().contains(keyColumn)) {
			throw new IllegalArgumentException(String.format("Key column: %s NOT found in csv: %s", keyColumn, csvPath));
		}
		HashBasedTable<String, String, String> ret = HashBasedTable.create();
		for (CSVRecord rec : csvContentTup.getRight()) {
			String rowKey = rec.get(keyColumn);
			for (String column : csvContentTup.getLeft()) {
				ret.put(rowKey, column, rec.get(column));
			}
		}
		return ret;
	}

	/**
	 *
	 * @param csvIn csvIn
	 * @param colName colName
	 * @return List&lt;String&gt;
	 * @throws IOException IOException
	 */
	public static List<String> readCsvColumn(InputStream csvIn, String colName) throws IOException {

		return readCsvColumn(csvIn, colName, "UTF-8");
	}

	/**
	 *
	 * @param csvIn csvIn
	 * @param colName colName
	 * @param enc enc
	 * @return List&lt;String&gt;
	 * @throws IOException IOException
	 */
	public static List<String> readCsvColumn(InputStream csvIn, String colName, String enc) throws IOException {

		List<String> ret = new ArrayList<>();
		CSVParser parser = CSVParser.parse(csvIn, Charset.forName(enc),
				CSVFormat.DEFAULT.withFirstRecordAsHeader());
		List<String> headerNames = parser.getHeaderNames();
		if (headerNames.contains(colName)) {
			List<CSVRecord> records = parser.getRecords();
			for (CSVRecord r : records) {
				ret.add(r.get(colName));
			}
		}

		return ret;
	}

	/**
	 * 
	 * @param csvPath csvPath
	 * @return Table
	 * @throws IOException IOException
	 */
	public static Table readTableFromCsv(Path csvPath) throws IOException {

		return readTableFromCsv(csvPath, "UTF-8");
	}

	/**
	 * 注意: csvIn 必须支持 mark/reset, 否则抛出 IOException:
	 *
	 * @param csvIn csvIn must support mark/reset
	 * @return Table
	 * @throws IOException IOException
	 */
	public static Table readTableFromCsv(InputStream csvIn) throws IOException {

		return readTableFromCsv(csvIn, "UTF-8");
	}

	/**
	 * 
	 * @param csvPath csvPath
	 * @param enc default: "UTF-8"
	 * @return Table
	 * @throws IOException IOException
	 */
	public static Table readTableFromCsv(Path csvPath, String enc) throws IOException {

		Map<String, Integer> csvHeaderMap = csvHeaderMap(csvPath, enc);
		ColumnType[] colTypes = new ColumnType[csvHeaderMap.size()];
		for (String e : csvHeaderMap.keySet()) {
			colTypes[csvHeaderMap.get(e)] = ColumnType.STRING;
		}
		Table ret = null;
		try (InputStreamReader reader = new InputStreamReader(new FileInputStream(csvPath.toFile()),
				Charset.forName(enc))) {
			CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("")
					.columnTypes(colTypes).build();

			ret = Table.read().usingOptions(options);
		}
		return ret;
	}

	/**
	 * 注意: csvIn 必须支持 mark/reset, 否则抛出 IOException:
	 *
	 * @param csvIn csvIn must support mark/reset
	 * @param enc default: "UTF-8"
	 * @return Table
	 * @throws IOException IOException
	 */
	public static Table readTableFromCsv(InputStream csvIn, String enc) throws IOException {

		if (csvIn.markSupported()) {
			Map<String, Integer> csvHeaderMap = csvHeaderMap(csvIn, enc);
			ColumnType[] colTypes = new ColumnType[csvHeaderMap.size()];
			for (String e : csvHeaderMap.keySet()) {
				colTypes[csvHeaderMap.get(e)] = ColumnType.STRING;
			}
			csvIn.reset();
			Table ret = null;
			InputStreamReader reader = new InputStreamReader(csvIn, Charset.forName(enc));

			CsvReadOptions options = CsvReadOptions.builder(reader).missingValueIndicator("")
					.columnTypes(colTypes).build();

			ret = Table.read().usingOptions(options);

			return ret;
		} else {
			throw new IOException(String.format("mark/reset not supported for InputStream: %s", csvIn));
		}
	}

	/**
	 * <pre>
	 * 检查 csv 数据行与 header 是否一致, 若仅最后一行不一致, 则忽略, 否则抛出异常;
	 * 返回 tuple &lt;是否一致, 忽略最后一行的 csv 转 String 内容&gt;
	 * </pre>
	 * 
	 * @param csv csv
	 * @return Pair&lt;Boolean, String&gt;
	 * @throws IOException IOException
	 */
	public static Pair<Boolean, String> csvToStringIfNotConsistent(File csv) throws IOException {

		return csvToStringIfNotConsistent(csv, CSV_FORMAT_DEFAULT, "UTF-8");
	}

	/**
	 * <pre>
	 * 检查 csv 数据行与 header 是否一致, 若仅最后一行不一致, 则忽略, 否则抛出异常;
	 * 返回 tuple &lt;是否一致, 忽略最后一行的 csv 转 String 内容&gt;
	 * </pre>
	 * 
	 * @param csv csv
	 * @param fmt fmt
	 * @return Pair&lt;Boolean, String&gt;
	 * @throws IOException IOException
	 */
	public static Pair<Boolean, String> csvToStringIfNotConsistent(File csv, CSVFormat fmt) throws IOException {

		return csvToStringIfNotConsistent(csv, fmt, "UTF-8");
	}

	/**
	 * <pre>
	 * 检查 csv 数据行与 header 是否一致, 若仅最后一行不一致, 则忽略, 否则抛出异常;
	 * 返回 tuple &lt;是否一致, 忽略最后一行的 csv 转 String 内容&gt;
	 * </pre>
	 * 
	 * @param csv csv
	 * @param fmt fmt
	 * @param enc enc
	 * @return Pair&lt;Boolean, String&gt;
	 */
	public static Pair<Boolean, String> csvToStringIfNotConsistent(File csv, CSVFormat fmt, String enc)
	        throws IOException {

		boolean isConsistent = true;
		try (CSVParser parser = CSVParser.parse(csv, Charset.forName(enc),
		        fmt.withFirstRecordAsHeader().withIgnoreEmptyLines());
		        StringWriter fixedCsvWriter = new StringWriter();
		        CSVPrinter printer = new CSVPrinter(fixedCsvWriter, fmt)) {

			printer.printRecord(parser.getHeaderNames());
			List<CSVRecord> records = parser.getRecords();
			for (int i = 0; i < records.size(); i++) {
				CSVRecord rec = records.get(i);
				if (rec.isConsistent()) {
					printer.printRecord(IteratorUtils.toArray(rec.iterator()));
				} else {
					if (i != records.size() - 1) {
						throw new IOException(String.format("inconsistent record (recordNum: %s): %s; columns: %s", i,
						        rec, parser.getHeaderNames()));
					} else {
						// ignore the last row:
						isConsistent = false;
					}
				}
			}
			if (isConsistent) {
				return Pair.of(false, null);
			} else {
				printer.flush();
				return Pair.of(true, fixedCsvWriter.toString());
			}
		}
	}

	/**
	 * 替代 {@link DataFrameWriter#csv(File) Table.write().csv()}
	 * ({@link NumberColumn#setPrintFormatter(tech.tablesaw.columns.numbers.NumberColumnFormatter)
	 * setPrintFormatter}), 支持自定义字段输出格式的 csv 导出.
	 * 
	 * @param table table
	 * @param csv csv
	 * @throws IOException IOException
	 */
	public static void toCsv(Table table, File csv) throws IOException {

		toCsv(table, csv, CSV_FORMAT_DEFAULT, false, "UTF-8");
	}

	/**
	 * 替代 {@link DataFrameWriter#csv(File) Table.write().csv()}
	 * ({@link NumberColumn#setPrintFormatter(tech.tablesaw.columns.numbers.NumberColumnFormatter)
	 * setPrintFormatter}), 支持自定义字段输出格式的 csv 导出.
	 * 
	 * @param table table
	 * @param csv csv
	 * @param fmt fmt
	 * @throws IOException IOException
	 */
	public static void toCsv(Table table, File csv, CSVFormat fmt) throws IOException {

		toCsv(table, csv, fmt, false, "UTF-8");
	}

	/**
	 * 替代 {@link DataFrameWriter#csv(File) Table.write().csv()}
	 * ({@link NumberColumn#setPrintFormatter(tech.tablesaw.columns.numbers.NumberColumnFormatter)
	 * setPrintFormatter}), 支持自定义字段输出格式的 csv 导出.
	 * 
	 * @param table table
	 * @param csv csv
	 * @param fmt fmt
	 * @param rewrite rewrite
	 * @throws IOException IOException
	 */
	public static void toCsv(Table table, File csv, CSVFormat fmt, boolean rewrite) throws IOException {

		toCsv(table, csv, fmt, rewrite, "UTF-8");
	}

	/**
	 * 替代 {@link DataFrameWriter#csv(File) Table.write().csv()}
	 * ({@link NumberColumn#setPrintFormatter(tech.tablesaw.columns.numbers.NumberColumnFormatter)
	 * setPrintFormatter}), 支持自定义字段输出格式的 csv 导出.
	 * 
	 * @param table table
	 * @param csv csv
	 * @param fmt fmt
	 * @param rewrite rewrite
	 * @param enc enc
	 * @throws IOException IOException
	 */
	public static void toCsv(Table table, File csv, CSVFormat fmt, boolean rewrite, String enc) throws IOException {

		if (!csv.exists() || rewrite) {
			try (CSVPrinter printer = new CSVPrinter(new FileWriterWithEncoding(csv, enc), fmt)) {
				// print csv header:
				Object[] colNames = new Object[table.columnCount()];
				for (int i = 0; i < table.columnCount(); i++) {
					colNames[i] = table.column(i).name();
				}
				printer.printRecord(colNames);
				// print each rows:
				for (int i = 0; i < table.rowCount(); i++) {
					Object[] row = new Object[table.columnCount()];
					for (int j = 0; j < table.columnCount(); j++) {
						row[j] = table.column(j).getString(i);
					}
					printer.printRecord(row);
				}
			} finally {

			}
		}
	}

	private CommonUtils() {

	}
}
